package com.hadoop.hbase.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * HBase连接需要在客户端DNS或者hosts文件中配置HBase中节点机器的名字
 */
public class Client {

    private static final String TABLE_NAME = "test_table";

    private static final String INFO = "info";

    private static final String NAME = "name";

    private static final String AGE = "age";

    private static final String ZOOKEEPERS = "172.16.1.113:2181,172.16.1.114:2181,172.16.1.115:2181";

    private static final String QUORUM = "hbase.zookeeper.quorum";


    public static void main(String[] args) throws IOException {
        //createTable();
        listTables();
        //addDataInTable();
        //readTable();
    }


    public static void readTable() throws IOException {
        //连接hbase集群
        Configuration configuration = HBaseConfiguration.create();
        //指定hbase的zk连接地址
        configuration.set(QUORUM, ZOOKEEPERS);
        Connection connection = ConnectionFactory.createConnection(configuration);
        TableName tableName = TableName.valueOf(TABLE_NAME);
        Table table = connection.getTable(tableName);
        Scan scan = new Scan();
        ResultScanner results = table.getScanner(scan);

        for (Result result : results) {
            String rowKey = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue(Bytes.toBytes(NAME), Bytes.toBytes(NAME)));
            String age = Bytes.toString(result.getValue(Bytes.toBytes(AGE), Bytes.toBytes(AGE)));
            System.out.println("rowKey : " + rowKey + " name : " + name + " age : " + age);
        }
        table.close();
    }


    public static void addDataInTable() throws IOException {
        //连接hbase集群
        Configuration configuration = HBaseConfiguration.create();
        //指定hbase的zk连接地址
        configuration.set(QUORUM, ZOOKEEPERS);
        Connection connection = ConnectionFactory.createConnection(configuration);
        //获取管理员对象
        Admin admin = connection.getAdmin();
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        /*boolean b = admin.tableExists(TableName.valueOf(TABLE_NAME));
        if (!b) {
            return;
        }*/
        for (int i = 0; i < 20; i++) {
            Put put = new Put(new Integer(i).toString().getBytes());
            put.addColumn(NAME.getBytes(), NAME.getBytes(), ("小明" + i).getBytes());
            put.addColumn(AGE.getBytes(), AGE.getBytes(), ("20" + i).getBytes());
            table.put(put);
        }

    }

    public static void createTable() throws IOException {
        //连接hbase集群
        Configuration configuration = HBaseConfiguration.create();
        //指定hbase的zk连接地址
        configuration.set(QUORUM, ZOOKEEPERS);
        Connection connection = ConnectionFactory.createConnection(configuration);
        //获取管理员对象
        Admin admin = connection.getAdmin();
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("default").build();
        admin.modifyNamespace(namespaceDescriptor);
        //通过管理员对象创建表
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
        //给我们的表添加列族，指定两个列族  f1   f2
        HColumnDescriptor f1 = new HColumnDescriptor(NAME);
        HColumnDescriptor f2 = new HColumnDescriptor(AGE);
        //将两个列族设置到  hTableDescriptor里面去
        hTableDescriptor.addFamily(f1);
        hTableDescriptor.addFamily(f2);
        //创建表
        admin.createTable(hTableDescriptor);
        admin.close();
        connection.close();
    }

    public static void listTables() throws IOException {
        //连接hbase集群
        Configuration configuration = HBaseConfiguration.create();
        //指定hbase的zk连接地址
        configuration.set(QUORUM, ZOOKEEPERS);
        Connection connection = ConnectionFactory.createConnection(configuration);
        //获取管理员对象
        Admin admin = connection.getAdmin();
        HTableDescriptor[] hTableDescriptors = admin.listTables();
        List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();

        for (HTableDescriptor hTableDescriptor : hTableDescriptors) {
            System.out.println(hTableDescriptor);
        }
        for (TableDescriptor tableDescriptor : tableDescriptors) {
            System.out.println(tableDescriptor);
        }
        admin.close();
        connection.close();
    }
}
